package org.javaboy;

import java.util.Scanner;
import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

public class FlowDemo {
    public static void main(String[] args) {
        class DataFilter extends SubmissionPublisher<String> implements Flow.Processor<String, String> {
            Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                //向生产者请求1条数据回来，生产者会发回来1条数据，然后调用onNext方法
                subscription.request(1);
            }

            @Override
            public void onNext(String item) {
                this.submit("[这是一条被处理过的数据]" + item);
                this.subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                this.subscription.cancel();
            }

            @Override
            public void onComplete() {
                this.close();
            }
        }
        //1. 定义消息生产者
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
        //2. 定义消息消费者
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {
            Flow.Subscription subscription;

            /**
             * 订阅消息，相当于消费者和生产者建立连接的时候调用
             * @param subscription a new subscription
             */
            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                System.out.println("连接建立成功");
                this.subscription = subscription;
                //向生产者请求1条数据回来，生产者会发回来1条数据，然后调用onNext方法
                subscription.request(1);
            }

            /**
             * 消费消息
             * 从生产者发送来的消息就是到达这里
             * 如果我们希望按照自己的消费节奏来，消费一条消息，消费完之后再去请求生产者发一条消息过来
             * @param item the item
             */
            @Override
            public void onNext(String item) {
                System.out.println("item = " + item);
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                this.subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                //消费消息失败
                this.subscription.cancel();
            }

            @Override
            public void onComplete() {
                System.out.println("数据接收完毕！");
            }
        };
        DataFilter dataFilter = new DataFilter();
        //dataFilter 作为消费者，设置给生产者
        publisher.subscribe(dataFilter);
        //dataFilter 作为生产者，给他设置消费者
        dataFilter.subscribe(subscriber);
        for (int i = 0; i < 1000; i++) {
            String item = "hello " + i;
            publisher.submit(item);
            System.out.println("publisher-->" + item);
        }
        publisher.close();
        new Scanner(System.in).next();
    }
}
